package cn.org.wangchangjiu.delay;

import lombok.extern.slf4j.Slf4j;

import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Classname DelayQueueMessageConsumer
 * @Description 延迟队列消费者
 * @Date 2022/9/14 10:28
 * @Created by wangchangjiu
 */
@Slf4j
public class DelayQueueMessageConsumer {

    private DelayQueueMessage delayQueueMessage;

    private KafkaSimpleMessageProducer kafkaSimpleMessageProducer;

    private String registerService;

    public DelayQueueMessageConsumer(DelayQueueMessage delayQueueMessage,
                                     KafkaSimpleMessageProducer kafkaSimpleMessageProducer,
                                     String registerService){
        this.delayQueueMessage = delayQueueMessage;
        this.kafkaSimpleMessageProducer = kafkaSimpleMessageProducer;
        this.registerService = registerService;
    }



    @PostConstruct
    public void consumerAndSendKafka() throws Exception {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        executorService.submit(() -> {
            String QUEUE_CODE = String.format(Constant.QUEUE_CODE, registerService);
            while (true) {
                Object obj = delayQueueMessage.getDelayQueue(QUEUE_CODE);
                if(obj != null){
                    RedisDelayMessage message = RedisDelayMessage.class.cast(obj);
                    kafkaSimpleMessageProducer.sendSimpleMessage(message.getTopic(), message.getValue());
                    log.info("--延迟队列中间件...topic:{}, 获取延迟消息：{}, 发送kafka消息队列中--", message.getTopic(), message.getValue());
                }
                Thread.sleep(500);
            }
        });
        log.info("(Redis延迟队列启动成功)");
    }
}
